"""
警告抑制工具

用于抑制第三方库的兼容性警告，统一管理项目中的所有警告过滤器
"""
import warnings
from typing import List, Tuple


class WarningSuppressor:
    """警告抑制器 - 统一管理所有警告过滤器"""
    
    def __init__(self):
        self.suppressed_warnings: List[Tuple[str, str, str]] = []
    
    def suppress_pydantic_langchain_warnings(self):
        """抑制 LangChain 与 Pydantic V2 的兼容性警告"""
        warning_filters = [
            # 抑制 Pydantic __fields__ 弃用警告
            ("ignore", ".*__fields__.*deprecated.*", "DeprecationWarning"),
            # 抑制 Pydantic model_fields 相关警告
            ("ignore", ".*model_fields.*deprecated.*", "DeprecationWarning"),
            # 抑制来自 langchain_core.tools.base 的警告
            ("ignore", ".*PydanticDeprecatedSince20.*", "DeprecationWarning"),
        ]
        
        for action, message, category in warning_filters:
            warnings.filterwarnings(action, message=message, category=DeprecationWarning)
            self.suppressed_warnings.append((action, message, category))
    
    def suppress_qdrant_warnings(self):
        """抑制 qdrant_client 相关警告"""
        warning_filters = [
            ("ignore", ".*", "DeprecationWarning", "qdrant_client"),
            ("ignore", ".*invalid escape sequence.*", "DeprecationWarning"),
        ]
        
        for filter_info in warning_filters:
            if len(filter_info) == 4:
                action, message, category, module = filter_info
                warnings.filterwarnings(action, message=message, category=DeprecationWarning, module=module)
            else:
                action, message, category = filter_info
                warnings.filterwarnings(action, message=message, category=DeprecationWarning)
            self.suppressed_warnings.append(filter_info)
    
    def suppress_mcp_warnings(self):
        """抑制 MCP 相关的异步和HTTP警告"""
        warning_filters = [
            # HTTP异步生成器相关警告
            ("ignore", ".*async generator ignored GeneratorExit.*", "RuntimeWarning"),
            ("ignore", ".*coroutine.*was never awaited.*", "RuntimeWarning"),
            ("ignore", ".*Attempted to exit cancel scope in a different task.*", "RuntimeWarning"),
            # 资源警告
            ("ignore", ".*", "ResourceWarning"),
        ]
        
        for action, message, category in warning_filters:
            if category == "ResourceWarning":
                warnings.filterwarnings(action, category=ResourceWarning)
            else:
                warnings.filterwarnings(action, message=message, category=RuntimeWarning)
            self.suppressed_warnings.append((action, message, category))
    
    def suppress_asyncio_warnings(self):
        """抑制 asyncio 相关的资源警告"""
        warning_filters = [
            # 未关闭的传输警告
            ("ignore", "unclosed transport.*", "ResourceWarning"),
            # 未关闭的事件循环警告
            ("ignore", "unclosed event loop.*", "ResourceWarning"),
            # I/O操作相关警告
            ("ignore", ".*I/O operation on closed.*", "RuntimeWarning"),
        ]
        
        for action, message, category in warning_filters:
            if category == "ResourceWarning":
                warnings.filterwarnings(action, message=message, category=ResourceWarning)
            else:
                warnings.filterwarnings(action, message=message, category=RuntimeWarning)
            self.suppressed_warnings.append((action, message, category))
    
    def suppress_all_warnings(self):
        """抑制所有已知的第三方库警告"""
        print("🔇 开始初始化警告抑制器...")
        
        # 抑制各类警告
        self.suppress_pydantic_langchain_warnings()
        self.suppress_qdrant_warnings() 
        self.suppress_mcp_warnings()
        self.suppress_asyncio_warnings()
        
        print(f"✅ 警告抑制器初始化完成，已抑制 {len(self.suppressed_warnings)} 类警告")
    
    def get_suppressed_count(self) -> int:
        """获取已抑制的警告数量"""
        return len(self.suppressed_warnings)
    
    def get_suppressed_details(self) -> List[Tuple[str, str, str]]:
        """获取已抑制的警告详情"""
        return self.suppressed_warnings.copy()
    
    def reset_warnings(self):
        """重置所有警告过滤器"""
        warnings.resetwarnings()
        self.suppressed_warnings.clear()
        print("🔄 警告过滤器已重置")


# 全局警告抑制器实例
warning_suppressor = WarningSuppressor()

def init_warning_suppression():
    """初始化警告抑制 - 项目入口点调用"""
    warning_suppressor.suppress_all_warnings()
    return warning_suppressor

def suppress_pydantic_warnings():
    """快捷方法：只抑制 Pydantic 相关警告"""
    warning_suppressor.suppress_pydantic_langchain_warnings()

def suppress_mcp_warnings():
    """快捷方法：只抑制 MCP 相关警告"""
    warning_suppressor.suppress_mcp_warnings()

def suppress_asyncio_warnings():
    """快捷方法：只抑制 asyncio 相关警告"""
    warning_suppressor.suppress_asyncio_warnings() 